package com.example.handler;

import com.example.auth.MqttAuthenticator;
import com.example.session.MqttSession;
import com.example.session.SessionManager;
import com.example.persistence.MessageStore;
import com.example.config.MqttServerConfig;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import java.util.List;
import java.util.stream.Collectors;
import io.netty.channel.Channel;

@Slf4j
@Component
@Scope("prototype")
public class MqttServerHandler extends ChannelInboundHandlerAdapter {
    
    private final MqttAuthenticator authenticator;
    private final SessionManager sessionManager;
    private final MessageStore messageStore;
    private final MqttServerConfig config;
    
    @Autowired
    public MqttServerHandler(MqttAuthenticator authenticator, 
                           SessionManager sessionManager,
                           MqttServerConfig config,
                           @Autowired(required = false) MessageStore messageStore) {
        this.authenticator = authenticator;
        this.sessionManager = sessionManager;
        this.config = config;
        this.messageStore = messageStore;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        String remoteAddress = ctx.channel().remoteAddress().toString();
        log.info("New connection from: {}", remoteAddress);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        String remoteAddress = ctx.channel().remoteAddress().toString();
        log.info("Connection closed from: {}", remoteAddress);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof MqttMessage) {
            MqttMessage mqttMessage = (MqttMessage) msg;
            String remoteAddress = ctx.channel().remoteAddress().toString();
            log.debug("Received MQTT message type: {} from: {}", mqttMessage.fixedHeader().messageType(), remoteAddress);
            
            switch (mqttMessage.fixedHeader().messageType()) {
                case CONNECT:
                    handleConnect(ctx, (MqttConnectMessage) mqttMessage);
                    break;
                case PUBLISH:
                    handlePublish(ctx, (MqttPublishMessage) mqttMessage);
                    break;
                case PUBACK:
                    handlePubAck(ctx, mqttMessage);
                    break;
                case SUBSCRIBE:
                    handleSubscribe(ctx, (MqttSubscribeMessage) mqttMessage);
                    break;
                case PINGREQ:
                    handlePingReq(ctx);
                    break;
                case DISCONNECT:
                    handleDisconnect(ctx);
                    break;
            }
        }
    }

    private void handleConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
        String clientId = msg.payload().clientIdentifier();
        String username = msg.payload().userName();
        String password = msg.payload().passwordInBytes() != null ? 
            new String(msg.payload().passwordInBytes()) : null;

        if (!authenticator.authenticate(username, password)) {
            ctx.writeAndFlush(MqttMessageBuilders.connAck()
                .returnCode(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD)
                .build());
            ctx.close();
            return;
        }

        MqttSession session = new MqttSession();
        session.setClientId(clientId);
        session.setChannel(ctx.channel());
        session.setCleanSession(msg.variableHeader().isCleanSession());
        
        sessionManager.addSession(clientId, session);

        // 如果不是清理会话，恢复之前的订阅
        if (!msg.variableHeader().isCleanSession()) {
            // 恢复会话状态
            restoreSession(clientId);
        }

        ctx.channel().attr(MqttSession.CLIENT_ID).set(clientId);

        ctx.writeAndFlush(MqttMessageBuilders.connAck()
            .returnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED)
            .sessionPresent(!msg.variableHeader().isCleanSession())
            .build());
    }

    private void handlePublish(ChannelHandlerContext ctx, MqttPublishMessage msg) {
        int qos = msg.fixedHeader().qosLevel().value();
        
        // 存储消息
        saveMessage(msg);

        // 处理不同QoS级别
        switch (qos) {
            case 0:
                // QoS 0: 最多发送一次，不需要确认
                break;
            case 1:
                // QoS 1: 至少发送一次，需要PUBACK
                ctx.writeAndFlush(MqttMessageBuilders.pubAck()
                    .packetId(msg.variableHeader().packetId())
                    .build());
                break;
            case 2:
                // QoS 2: 确保只收到一次，需要PUBREC
                MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0);
                MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(msg.variableHeader().packetId());
                ctx.writeAndFlush(new MqttMessage(fixedHeader, variableHeader));
                break;
        }

        // 转发消息给订阅者
        forwardMessage(msg);
    }

    private void saveMessage(MqttPublishMessage msg) {
        // 只有在启用MySQL持久化且消息符合保存条件时才保存
        if (config.getPersistence().getMysql().isEnabled() && messageStore != null) {
            if (msg.fixedHeader().qosLevel().value() > 0 || 
                (msg.fixedHeader().qosLevel().value() == 0 && config.getPersistence().getMysql().isSaveQos0())) {
                com.example.persistence.MqttMessage message = new com.example.persistence.MqttMessage();
                message.setTopic(msg.variableHeader().topicName());
                message.setQos(msg.fixedHeader().qosLevel().value());
                message.setRetained(msg.fixedHeader().isRetain());
                message.setPayload(msg.payload().array());
                messageStore.save(message);
            }
        }
    }

    private void forwardMessage(MqttPublishMessage msg) {
        // 实现消息转发逻辑
    }

    private void restoreSession(String clientId) {
        // 恢复之前的会话状态
        messageStore.findByClientIdAndQosGreaterThan(clientId, 0)
            .forEach(this::resendMessage);
    }

    private void resendMessage(com.example.persistence.MqttMessage message) {
        // 实现消息重发逻辑
    }

    private void handlePubAck(ChannelHandlerContext ctx, MqttMessage msg) {
        // QoS 1消息确认处理
        int messageId = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
        // 可以在这里添加消息确认的处理逻辑
    }

    private void handleSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) {
        int messageId = msg.variableHeader().messageId();
        List<MqttTopicSubscription> subscriptions = msg.payload().topicSubscriptions();
        
        // 处理订阅请求
        MqttMessageBuilders.SubAckBuilder subAckBuilder = MqttMessageBuilders.subAck()
            .packetId(messageId);
            
        subscriptions.forEach(sub -> 
            subAckBuilder.addGrantedQos(sub.qualityOfService()));
            
        ctx.writeAndFlush(subAckBuilder.build());
    }

    private void handlePingReq(ChannelHandlerContext ctx) {
        // 回复 PINGRESP
        ctx.writeAndFlush(new MqttMessage(
            new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0)
        ));
    }

    private void handleDisconnect(ChannelHandlerContext ctx) {
        // 清理会话
        String clientId = ctx.channel().attr(MqttSession.CLIENT_ID).get();
        if (clientId != null) {
            sessionManager.removeSession(clientId);
        }
        ctx.close();
    }
} 